﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Text;

namespace Receive
{
    class Program
    {
        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "192.168.6.88", Port = 5672, UserName = "test", Password = "123456" };
            //var factory_uri = new ConnectionFactory { Uri = new Uri("amqp://test:123456@192.168.6.88:5672/") };
            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {

                    Demo1(channel);

                    Demo2(channel);

                    Demo3(channel);

                    Demo4(channel);
                    Demo4_1(channel);
                    Demo4_2(channel);

                    Demo5(channel);
                    Demo5_1(channel);
                    Demo5_2(channel);

                    Demo6(channel);

                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadKey();
                }
            }
        }

        /// <summary>
        /// 基础demo
        /// </summary>
        /// <param name="channel"></param>
        static void Demo1(IModel channel)
        {
            string queue = string.Format("MQ.{0}.Hello", "Demo1");
            channel.QueueDeclare(queue: queue, durable: false, exclusive: false, autoDelete: false, arguments: null);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {

                var body = ea.Body.ToArray();
                string message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [1] Received {0}", message);

                channel.BasicAck(ea.DeliveryTag, false); //手动Ack：用来确认消息已经被消费完成了
            };
            channel.BasicConsume(queue, false, consumer);    //开启消费者与通道、队列关联

            //channel.BasicConsume(queue, true, consumer);//开启消费者与通道、队列关联

        }
        /// <summary>
        /// 持久化
        /// </summary>
        /// <param name="channel"></param>
        static void Demo2(IModel channel)
        {
            string queue = string.Format("MQ.{0}.Hello", "Demo2");
            channel.QueueDeclare(queue: queue,
                                 durable: true,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: null);

            channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

            Console.WriteLine(" [*] Waiting for messages.");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (sender, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [2] Received {0}", message);

                int dots = message.Split('.').Length - 1;
                System.Threading.Thread.Sleep(dots * 1000);

                Console.WriteLine(" [2] Done");

                // Note: it is possible to access the channel via
                //       ((EventingBasicConsumer)sender).Model here
                channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            };
            channel.BasicConsume(queue: queue,
                                 autoAck: false,
                                 consumer: consumer);
        }
        /// <summary>
        /// Publish/Subscribe(fanout 模式)
        /// 一个消息可以到达多个bind的队列
        /// </summary>
        /// <param name="channel"></param>
        static void Demo3(IModel channel)
        {
            string exchange = string.Format("Ex.{0}.Hello", "Demo3");

            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Fanout);
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                exchange: exchange,
                routingKey: "");

            Console.WriteLine(" {0}[*] Waiting ...", queueName);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [3] Received {0}", message);
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
        }
        /// <summary>
        /// Routing 模式
        /// </summary>
        /// <param name="channel"></param>
        static void Demo4(IModel channel)
        {
            string exchange = string.Format("Ex.{0}.Hello", "Demo4");
            string routingKey = string.Format("{0}.log", "info");

            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Direct);
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                exchange: exchange,
                routingKey: routingKey);

            Console.WriteLine(" {0}[*] Waiting {1} ...", queueName, routingKey);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [4] Received {1}:{0}", message, ea.RoutingKey);
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
        }
        /// <summary>
        /// Routing 模式
        /// </summary>
        /// <param name="channel"></param>
        static void Demo4_1(IModel channel)
        {
            string exchange = string.Format("Ex.{0}.Hello", "Demo4");
            string routingKey = string.Format("{0}.log", "error");

            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Direct);
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                exchange: exchange,
                routingKey: routingKey);

            Console.WriteLine(" {0}[*] Waiting {1} ...", queueName, routingKey);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [4.1] Received {1}:{0}", message, ea.RoutingKey);
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
        }
        /// <summary>
        /// Routing 模式,二次订阅error
        /// </summary>
        /// <param name="channel"></param>
        static void Demo4_2(IModel channel)
        {
            string exchange = string.Format("Ex.{0}.Hello", "Demo4");
            string routingKey = string.Format("{0}.log", "error");

            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Direct);
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                exchange: exchange,
                routingKey: routingKey);

            Console.WriteLine(" {0}[*] Waiting {1} ...", queueName, routingKey);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [4.2] Received second {1}:{0}", message, ea.RoutingKey);
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
        }
        /// <summary>
        /// Topics 模式
        /// </summary>
        /// <param name="channel"></param>
        static void Demo5(IModel channel)
        {
            string exchange = string.Format("Ex.{0}.Hello", "Demo5");
            string routingKey = string.Format("{0}.info", "*");

            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Topic);
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                exchange: exchange,
                routingKey: routingKey);

            Console.WriteLine(" {0}[*] Waiting {1} ...", queueName, routingKey);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [5] Received {1}:{0}", message, ea.RoutingKey);
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
        }
        /// <summary>
        /// Topics 模式
        /// </summary>
        /// <param name="channel"></param>
        static void Demo5_1(IModel channel)
        {
            string exchange = string.Format("Ex.{0}.Hello", "Demo5");
            string routingKey = string.Format("info.{0}", "*");

            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Topic);
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                exchange: exchange,
                routingKey: routingKey);

            Console.WriteLine(" {0}[*] Waiting {1} ...", queueName, routingKey);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [5.1] Received {1}:{0}", message, ea.RoutingKey);
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
        }
        /// <summary>
        /// Topics 模式，全部监听
        /// </summary>
        /// <param name="channel"></param>
        static void Demo5_2(IModel channel)
        {
            string exchange = string.Format("Ex.{0}.Hello", "Demo5");
            string routingKey = string.Format("{0}", "#");

            channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Topic);
            var queueName = channel.QueueDeclare().QueueName;
            channel.QueueBind(queue: queueName,
                exchange: exchange,
                routingKey: routingKey);

            Console.WriteLine(" {0}[*] Waiting {1} ...", queueName, routingKey);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine(" [5.2] Received all {1}:{0}", message, ea.RoutingKey);
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
        }
        
        /// <summary>
        /// RPC
        /// </summary>
        /// <param name="channel"></param>
        static void Demo6(IModel channel)
        {
            var rpcClient = new RpcClient(channel);

            Console.WriteLine(" [x] Requesting fib(30)");
            var response = rpcClient.Call("30");

            Console.WriteLine(" [.] Got '{0}'", response);

        }
    }


    public class RpcClient
    {
        private readonly IModel channel;
        private readonly string replyQueueName;
        private readonly EventingBasicConsumer consumer;
        private readonly BlockingCollection<string> respQueue = new BlockingCollection<string>();
        private readonly IBasicProperties props;

        public RpcClient(IModel channel)
        {
            this.channel = channel;
            string queue = string.Format("MQ.{0}.Hello", "Demo6");
            channel.QueueDeclare(queue: queue, durable: false, exclusive: false, autoDelete: false, arguments: null);
            replyQueueName = channel.QueueDeclare().QueueName;
            Console.WriteLine(" [6] replyQueueName:{0}", replyQueueName);
            consumer = new EventingBasicConsumer(channel);

            props = channel.CreateBasicProperties();
            var correlationId = Guid.NewGuid().ToString();
            props.CorrelationId = correlationId;
            props.ReplyTo = replyQueueName;

            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var response = Encoding.UTF8.GetString(body.ToArray());
                if (ea.BasicProperties.CorrelationId == correlationId)
                {
                    respQueue.Add(response);
                }
            };
        }

        public string Call(string message)
        {
            string routingKey = string.Format("MQ.{0}.{1}", "Demo6", "Demo6");

            var messageBytes = Encoding.UTF8.GetBytes(message);
            channel.BasicPublish(exchange: "", routingKey: routingKey, basicProperties: props, body: messageBytes);

            channel.BasicConsume(consumer: consumer, queue: replyQueueName, autoAck: true);

            return respQueue.Take();
        }
    }

}
